package com.cqs.learning.spring.cloud.producer.bean.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cqs.learning.spring.cloud.producer.bean.dto.HousePriceChannel;
import com.cqs.learning.spring.cloud.producer.bean.po.HousePrice;
import com.cqs.learning.spring.cloud.producer.service.HousePriceService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;

import javax.annotation.Resource;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 *
 * //TODO kafka改成手动提交比较好，防止重启的时候，没有消费完的消息丢失。
 * @author li
 * @Date 5/31/20 11:25 PM
 */
@EnableBinding(value = HousePriceChannel.class)
@Slf4j
public class HousePriceStreamConsumer {


    @Resource
    private HousePriceService housePriceService;

    private final static int MAX_CAPACITY = 1000;
    private final static BlockingQueue<Runnable> housePriceQueue = new ArrayBlockingQueue<>(MAX_CAPACITY);

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(16, 16, 10, TimeUnit.SECONDS,
            housePriceQueue, r -> new Thread(r, "house-price-"), new ThreadPoolExecutor.AbortPolicy());
    private Random rand = new Random();

    /**
     * 消費Kafka里面的消息
     *
     * @param message
     */
    @StreamListener(HousePriceChannel.INPUT)
    public void consumerHousePriceMsg(Message<JSONObject> message) {
        JSONObject payload = message.getPayload();
        //log.info("订阅House Price消息：{},housePriceService:{}", payload, housePriceService);
        if (housePriceQueue.size() > (MAX_CAPACITY * 0.75)) {
            log.warn("队列中任务数量过多,进入休眠状态");
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //提交任务
        executor.submit(new HousePriceConsumerTask(payload));

    }


    class HousePriceConsumerTask implements Runnable {
        private final JSONObject message;
        HousePriceConsumerTask(JSONObject message) {
            this.message = message;
        }
        @Override
        public void run() {
            //1. 转化为对象
            HousePrice housePrice = JSONObject.toJavaObject(message, HousePrice.class);
            housePrice.setTui(message.getString("Transaction unique identifier"));
            long start = System.currentTimeMillis();
            if (housePrice.getTransferDate() == null) {
                log.warn("不合法的数据{}:", message);
                return;
            }
            boolean save = housePriceService.save(housePrice);
            long cost = System.currentTimeMillis() - start;
            if (!save) {
                log.warn("保存失败:{},cost:{}", message, cost);
            } else {
                if (rand.nextInt(100) < 10) {
                    log.info("保存成功:id{},cost:{} 毫秒", housePrice.getId(), cost);
                }
            }
        }
    }



    public static void main(String[] args) {
        String s = "{\"PropertyType\":\"F\",\"District\":\"TORBAY\",\"@version\":\"1\",\"PAON\":\"VILLA PARADISO\",\"type\":\"csv_log\",\"path\":\"/home/li/workspace/dataset/HousePriceDataEnglandWales1995to2014/pp-1995.csv/pp-1995.csv\",\"Transaction unique identifier\":\"{A42E2F04-2538-4A25-94C5-49E29C6C8FA8}\",\"Duration\":\"L\",\"SAON\":\"FLAT 10\",\"TransferDate\":\"1995-01-31 00:00\",\"ppdCategory\":\"A\",\"Street\":\"HIGHER WARBERRY ROAD\",\"RecordStatus\":\"A\",\"OldOrNew\":\"N\",\"Locality\":\"TORQUAY\",\"host\":\"cqs\",\"@timestamp\":\"2020-06-01T16:25:18.379Z\",\"Price\":\"18500\",\"TownOrCity\":\"TORQUAY\",\"County\":\"TORBAY\",\"message\":\"\\\"{A42E2F04-2538-4A25-94C5-49E29C6C8FA8}\\\",\\\"18500\\\",\\\"1995-01-31 00:00\\\",\\\"TQ1 1RY\\\",\\\"F\\\",\\\"N\\\",\\\"L\\\",\\\"VILLA PARADISO\\\",\\\"FLAT 10\\\",\\\"HIGHER WARBERRY ROAD\\\",\\\"TORQUAY\\\",\\\"TORQUAY\\\",\\\"TORBAY\\\",\\\"TORBAY\\\",\\\"A\\\",\\\"A\\\"\\r\",\"Postcode\":\"TQ1 1RY\"}";
        JSONObject jsonObject = JSON.parseObject(s);
        HousePrice housePrice = JSON.parseObject(s, HousePrice.class);
        housePrice.setTui(jsonObject.getString("Transaction unique identifier"));
        System.out.println(JSON.toJSON(housePrice));
    }

}